package com.fowo.api.component;

import cn.hutool.extra.spring.SpringUtil;
import com.fowo.api.service.impl.SupplierFlowNodeServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 程序运行完后执行消费
 */
@Component
@Slf4j
public class StartInit implements ApplicationRunner {
    @Resource
    RedissonClient redissonClient;

    /**
     * 启动项目开启线程轮询
     *
     * @param args
     * @throws Exception
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 带有定时功能的线程池
        Environment env = SpringUtil.getApplicationContext().getEnvironment();
        String ip = InetAddress.getLocalHost().getHostAddress();
        String port = env.getProperty("spring.servlet.port");
        log.info("\n----------------------------------------------------------\n\t" +
                "Application is running! Access URLs:\n\t" +
                "External: \thttp://" + ip + ":" + port + "/\n\t" +
                "----------------------------------------------------------");
        //监听延迟队列
        SpringUtil.getBean(SupplierFlowNodeServiceImpl.class).invokeEndTimeCallback();
    }

    /**
     * 任务回调监听
     *
     * @param <T>
     */
    public abstract static class TaskEventListener<T> {
        public abstract void invoke(T t);
    }

    /**
     * 添加队列
     *
     * @param t        DTO传输类
     * @param delay    时间数量
     * @param timeUnit 时间单位
     * @param <T>      泛型
     */
    public <T> void addQueue(T t, long delay, TimeUnit timeUnit) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(t.getClass().getName());
        RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        delayedQueue.offer(t, delay, timeUnit);
        delayedQueue.destroy();
    }

    /**
     * 获取队列
     *
     * @param zClass            DTO泛型
     * @param taskEventListener 任务回调监听
     * @param <T>               泛型
     * @return
     */
    public <T> void getQueue(Class zClass, StartInit.TaskEventListener taskEventListener) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(zClass.getName());
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
        scheduler.scheduleWithFixedDelay(() -> {
            try {
                T t = blockingFairQueue.take();
                taskEventListener.invoke(t);
            } catch (InterruptedException ex) {
                throw new RuntimeException(ex);
            }
        }, 1, 2, TimeUnit.SECONDS);//1秒初始化之后执行一次，以后每3秒执行一次（频率可以适当调节）
    }
}